package com.huitone.smspfm.task;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import com.huitone.smspfm.logger.ExtLogger;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.core.script.RedisScript;

import com.bin.common.utils.DateTimeUtils;
import com.huitone.smspfm.core.config.AppStaticConfigs;
import com.huitone.smspfm.core.filter.AddBlackUserFilterTask;
import com.huitone.smspfm.core.filter.MoDesitinationMatchFilterTask;
import com.huitone.smspfm.redis.model.SmsEntity;
import com.huitone.smspfm.redis.pipeline.PipelineOpts.ABExcuteCommands;

/**   
* @Title: MoSmsProcessingManager.java 
* @Package com.huitone.smspfm.module 
* @Description: 上行短信处理管理器
* @author ybliang   
* @date 2017年3月15日 下午3:52:28 
* @version V1.0   
*/
public class MoSmsProcessingTask extends Thread {
	
	private static ExtLogger logger = ExtLogger.create(MoSmsProcessingTask.class);
	
	private ExecutorService blockingPool = new ThreadPoolExecutor(AppStaticConfigs.mosmsExcutePoolSize, 
			AppStaticConfigs.mosmsExcutePoolSize, 1L, TimeUnit.MINUTES, new ArrayBlockingQueue<>(AppStaticConfigs.mosmsExcutePoolSize * 2), new RejectedExecutionHandler() {
		
		@Override
		public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
			boolean offer = false;
			while(!executor.isShutdown() && !offer) {
				offer = executor.getQueue().offer(r);
				if (!offer) {
					try {
						TimeUnit.MILLISECONDS.sleep(10);
					} catch (InterruptedException e) {}
				}
			}
		}
	});
	
	private boolean toSleep = false;
	
	private long sleepTime = 1000;
	
	private static final String CurrMoSpeedS = "CurrMoSpeedS";
	private static final String CurrMoSpeedM = "CurrMoSpeedM";
	
	private static final byte[] CURR_MO_SPEEDS_B = AppStaticConfigs.stringRedisSerializer.serialize(CurrMoSpeedS);
	
	private static final byte[] CURR_MO_SPEEDM_B = AppStaticConfigs.stringRedisSerializer.serialize(CurrMoSpeedM);
	
	private byte[] currMoSpeedS_List = AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.RunningParamsHash + ":" + CurrMoSpeedS);
	private byte[] currMoSpeedM_List = AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.RunningParamsHash + ":" + CurrMoSpeedM);
	
	private long currSpeedS = 0;
	
	private long currSpeedM = 0;
	
	private final Object slock = new Object();
	
	private final Object mlock = new Object();
	
	private ArrayBlockingQueue<Map<byte[],byte[]>> speeds = new ArrayBlockingQueue<Map<byte[],byte[]>>(1000);
	
	private boolean keepGoing = true;
	
	@Override
	public void run() {
		
		if (AppStaticConfigs.speedStatisticsOn) {
			
			startUpdateSpeedTasks();
			
			startSpeedWriteTasks();
			
		}
		
		@SuppressWarnings("rawtypes")
		RedisScript<List> script = new DefaultRedisScript<List>(AppStaticConfigs.luaGetMoSms, List.class);
		
		List<String> keys = new ArrayList<>();
		keys.add(AppStaticConfigs.iSMMoGateList);
		keys.add(AppStaticConfigs.pSMCacheHash);
		
		while(isKeepGoing() && !this.isInterrupted()) {
			
			blockingPool.execute(new Runnable() {
				
				@Override
				public void run() {
					
					try {
						
						List<List<String>> entities = AppStaticConfigs.scriptOpts3.getSMCacheAndHash(script, keys, AppStaticConfigs.mosmsBatchSizePerTime + "");
						
						List<SmsEntity> lists = new ArrayList<>();
						for (int i = 0 ; i < entities.size() ; i ++) {
							List<String> list = entities.get(i);
							SmsEntity smsEntity = SmsEntity.parse(list);
							lists.add(smsEntity);
						}
						
						if (!lists.isEmpty()) {
							
							@SuppressWarnings("unchecked")
							List<SmsEntity>[] callback = new List[1];
							
							AddBlackUserFilterTask addBlackUserFilterTask = new AddBlackUserFilterTask(lists, callback, 0);
							addBlackUserFilterTask.start();
							addBlackUserFilterTask.join();
							
							List<SmsEntity> list = callback[0];
							
							if (!list.isEmpty()) {
								MoDesitinationMatchFilterTask matchFilterTask = new MoDesitinationMatchFilterTask(list);
								matchFilterTask.start();
								matchFilterTask.join();
							}
							
							if (AppStaticConfigs.speedStatisticsOn) {
								int size = lists.size();
								increaseCurrSpeedS(size);
								increaseCurrSpeedM(size);
							}
							
							toSleep = false;
						} else {
							toSleep = true;
						}
					} catch (Exception e) {
						logger.error("上行短信处理过程发生错误:" + e.getMessage());
					}
				}
			});
			
			try {
				if (toSleep) {
					Thread.sleep(sleepTime);
				}
			} catch (InterruptedException e) {}
		}
		
		if (blockingPool != null && !blockingPool.isShutdown()) {
			blockingPool.shutdown();
		}
		
		logger.info("短信平台上行短信处理服务已经关闭");
	}
	
	private void startUpdateSpeedTasks() {
		
		new Thread(new Runnable() {
			
			@Override
			public void run() {
				
				while(blockingPool != null && !blockingPool.isShutdown()) {
					
					long speed = getAndResetCurrSpeedS();
					
					speeds.offer(Collections.singletonMap(CURR_MO_SPEEDS_B, AppStaticConfigs.stringRedisSerializer.serialize(speed + ":" + DateTimeUtils.dateToStr2(null))));
					
					logger.debug("上行短信处理速度(条/s):" + speed);
					
					try {
						TimeUnit.MILLISECONDS.sleep(1000);
					} catch (InterruptedException e) {}
				}
			}
		}).start();
		
		new Thread(new Runnable() {
			
			@Override
			public void run() {
				
				int count = 0;
				
				while(blockingPool != null && !blockingPool.isShutdown()) {
					
					if (count >= 60) {
						
						long speed = getAndResetCurrSpeedM();
						
						speeds.offer(Collections.singletonMap(CURR_MO_SPEEDM_B, AppStaticConfigs.stringRedisSerializer.serialize(speed + ":" + DateTimeUtils.dateToStr2(null))));
						
						logger.debug("上行短信处理速度(条/m):" + speed);
						
						count = 0;
						
					} else {
						count ++;
					}
					
					try {
						TimeUnit.MILLISECONDS.sleep(1000);
					} catch (InterruptedException e) {}
				}
			}
		}).start();
	}
	
	private void startSpeedWriteTasks() {
		
		new Thread(new Runnable() {
			
			@Override
			public void run() {
				
				while (blockingPool != null && !blockingPool.isShutdown()) {
					
					Map<byte[],byte[]> map = speeds.poll();
					if (map != null && !map.isEmpty()) {
						AppStaticConfigs.pipelineOpts2.save(new ABExcuteCommands() {

							@Override
							public void excute(RedisConnection connection) {
								connection.hMSet(AppStaticConfigs.RunningParamsHashB, map);
								// 每秒的速率队列
								byte[] bytes = map.get(CURR_MO_SPEEDS_B);
								if (bytes != null) {
									connection.lPush(currMoSpeedS_List, bytes); 
									connection.lTrim(currMoSpeedS_List, 0, AppStaticConfigs.speedSLength - 1);
								}
								// 每分的速率队列
								byte[] bytem = map.get(CURR_MO_SPEEDM_B);
								if (bytem != null) {
									connection.lPush(currMoSpeedM_List, bytem); 
									connection.lTrim(currMoSpeedM_List, 0, AppStaticConfigs.speedMLength - 1);
								}
							}
						});
					} else {
						try {
							TimeUnit.MILLISECONDS.sleep(100);
						} catch (InterruptedException e) {}
					}
				}
			}
		}).start();
	}

	public void shutdown() {
		if (blockingPool != null) {
			blockingPool.shutdown();
		}
	}

	public boolean isKeepGoing() {
		return keepGoing;
	}

	public void setKeepGoing(boolean keepGoing) {
		this.keepGoing = keepGoing;
	}

	@Override
	public void destroy() {
		this.setKeepGoing(false);
		this.interrupt();
	}
	
	private long getAndResetCurrSpeedS() {
		long result = 0;
		synchronized (slock) {
			result = currSpeedS;
			currSpeedS = 0;
		}
		return result;
	}

	private void increaseCurrSpeedS(long count) {
		synchronized (slock) {
			this.currSpeedS += count;
		}
	}

	public long getCurrSpeedM() {
		return currSpeedM;
	}
	
	private long getAndResetCurrSpeedM() {
		long result = 0;
		synchronized (mlock) {
			result = currSpeedM;
			currSpeedM = 0;
		}
		return result;
	}

	private void increaseCurrSpeedM(long count) {
		synchronized (mlock) {
			this.currSpeedM += count;
		}
	}
}
